package com.feishi.core.util.queue.cxq;


import com.feishi.core.util.queue.Customer;
import com.feishi.core.util.queue.feishi.Msg;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.Thread.sleep;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: caixq
 * Date: 2018-09-05
 * Time: 下午4:09
 */
public class CxqExecutor<T,R> {
    private final ExecutorService pool;
    LinkedBlockingQueue<Msg<T, R>> queue=new LinkedBlockingQueue<>();
    Customer customer;
    private int status=0;
    int index;
    public void push(Msg<T,R> msg) {
        queue.offer(msg);
//        try {
//            queue.put(msg);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }

    public CxqExecutor(Customer customer, ExecutorService pool) {
        this.customer = customer;
        this.pool=pool;
    }

    public void execute() {
        if(status!=0){
            return;
        }
        pool.execute(()->{
//            while (true){
//
//                try {
//                    Msg<T, R> msg = queue.take();
//                    customer.execute(msg,index++,queue.isEmpty());
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            }
            while (!queue.isEmpty()){
                status=1;
                Msg<T, R> msg = queue.poll();
                try {
                    if (msg!=null)
                    customer.execute(msg,index++,queue.isEmpty());
                } catch (Exception e) {
                    e.printStackTrace();
                }
                status=2;
            }
            status=0;
        });
//        status=1;
    }
    public static void main(String[] args){
        AtomicInteger num = new AtomicInteger();
        CxqProducer<Integer,Integer> producer=new CxqProducer<>((integerMsg, sequence, endOfBatch) -> {
            num.getAndIncrement();
        });
        int count = (int) 1e8;
        for (int i = 0; i < count; i++) {
            producer.push(new Msg<>(i%20,i));
        }
        try {
            sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(num);
        System.out.println(count-num.get());
    }
}
